package git.soulbgm;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * TCP客户端
 *
 * @author SoulBGM
 * @date 2022/04/19 23:44
 */
public class TcpClient {

    private final static LogUtil log = new LogUtil(TcpClient.class);

    /**
     * 等待启动的超时时间 单位纳秒
     */
    private static final long TIMEOUT_TIME = 800 * 1000000L;

    private Bootstrap bootstrap;

    /**
     * 数据进入处理器
     */
    private ChannelInboundHandlerAdapter handle;

    /**
     * TCP传输数据的线程数
     */
    private int tcpWorkerThreadNumber;

    /**
     * 管理传输数据线程组
     */
    private EventLoopGroup workGroup;

    private final static Map<String, ChannelFuture> CONNECT_POOL = new ConcurrentHashMap<>();

    public static ChannelFuture get(String ip, int port) {
        return CONNECT_POOL.get(ip + ":" + port);
    }

    public static void add(String ip, int port, ChannelFuture future) {
        CONNECT_POOL.put(ip + ":" + port, future);
    }

    public static void remove(String ip, int port) {
        CONNECT_POOL.remove(ip + ":" + port);
    }

    private static TcpClient client;

    private TcpClient() {
    }

    public static TcpClient getClient() {
        if (client == null) {
            client = new TcpClient();
        }
        return client;
    }

    public void start(ChannelInboundHandlerAdapter handle) {
        start(handle, 3);
    }

    public void start(ChannelInboundHandlerAdapter handle, int tcpWorkerThreadNumber) {
        this.handle = handle;
        this.tcpWorkerThreadNumber = tcpWorkerThreadNumber;
        initConfig();
    }

    /**
     * 初始化配置
     */
    private void initConfig() {
        ByteBuf delimiter = Unpooled.copiedBuffer(new byte[]{120, 120, 120, 120, -120, -120, -120, -120});
        workGroup = new NioEventLoopGroup(tcpWorkerThreadNumber);
        bootstrap = new Bootstrap()
                .group(workGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline()
                                /*
                                    第一个参数 30 表示读操作空闲时间
                                    第二个参数 30 表示写操作空闲时间
                                    第三个参数 60 表示读写操作空闲时间
                                    第四个参数 单位/秒
                                 */
                                .addLast(new IdleStateHandler(30, 60, 60, TimeUnit.SECONDS))
                                .addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, delimiter))
                                .addLast(handle)
                                .addLast(new StringEncoder());
                    }
                });
    }

    /**
     * 连接${host}:${port}
     *
     * @param host IP
     * @param port 端口
     * @return 通道异步任务结果
     */
    public ChannelFuture connect(final String host, final int port) {
        return connect(host, port, getChannelFutureListener(host, port));
    }

    /**
     * 连接${host}:${port}
     *
     * @param host IP
     * @param port 端口
     * @return 通道异步任务结果
     */
    public ChannelFuture connect(final String host, final int port, ChannelFutureListener channelFutureListener) {
        ChannelFuture future = bootstrap.connect(host, port);
        future.addListener(channelFutureListener);
        // 系统当前时间 纳秒
        long nowTime = System.nanoTime();
        while ((System.nanoTime() - nowTime) < TIMEOUT_TIME) {
            if (future.channel() != null) {
                break;
            }
            sleep(2, 400);
        }
        return future;
    }

    public ChannelFutureListener getChannelFutureListener(String host, int port) {
        return new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    add(host, port, future);
                    log.info("TCP客户端连接 {}:{} 服务器成功", host, port);
                } else {
                    future.channel().eventLoop().schedule(getReconnectRunnable(host, port), 3L, TimeUnit.SECONDS);
                }
            }
        };
    }

    public Runnable getReconnectRunnable(String ip, int port) {
        return getReconnectRunnable(ip, port, getChannelFutureListener(ip, port));
    }

    public Runnable getReconnectRunnable(String ip, int port, ChannelFutureListener channelFutureListener) {
        return new Runnable() {
            @Override
            public void run() {
                log.warn("IP:{}, 端口:{} 连接断开，正在连接重试~~", ip, port);
                try {
                    connect(ip, port, channelFutureListener);
                } catch (Exception e) {
                    log.error("TCP客户端重新连接出现异常", e);
                }
            }
        };
    }

    /**
     * 线程睡眠一段时间
     *
     * @param millis 毫秒数
     * @param nanos  纳秒数
     */
    private void sleep(long millis, int nanos) {
        try {
            Thread.sleep(millis, nanos);
        } catch (InterruptedException e) {
            log.info("获得TCP客户端休眠被中断", e);
            // 恢复中断状态
            Thread.currentThread().interrupt();
        }
    }

    public void info(String s, String host, int port) {
        log.info(s, host, port);
    }

    public void warn(String s, String ip, int port) {
        log.warn(s, ip, port);
    }
}
